"""
The self-pipe trick for Unix signal handler:
https://github.com/benoitc/gunicorn/blob/master/gunicorn/arbiter.py

>>> _test_signal_pipe()
"""
import os
import errno
import fcntl
import select


def close_on_exec(fd):
    flags = fcntl.fcntl(fd, fcntl.F_GETFD)
    flags |= fcntl.FD_CLOEXEC
    fcntl.fcntl(fd, fcntl.F_SETFD, flags)


def set_non_blocking(fd):
    flags = fcntl.fcntl(fd, fcntl.F_GETFL) | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)


class SignalPipe:
    def __init__(self):
        self._rfile, self._wfile = os.pipe()
        for fd in [self._rfile, self._wfile]:
            set_non_blocking(fd)
            close_on_exec(fd)
        self._signal_queue = []

    def handler(self, sig: int, frame):
        if len(self._signal_queue) < 5:
            self._signal_queue.append(sig)
            self._wakeup()

    def wait(self, timeout=None) -> int:
        """
        timeout (int|float): timeout in seconds;
            If it is absent or None, the call will never timeout.
        """
        while True:
            sig = self._pop_signal()
            if sig is not None:
                return sig
            self._sleep(timeout=timeout)
            if timeout is not None:
                sig = self._pop_signal()
                return sig

    def close(self):
        os.close(self._rfile)
        os.close(self._wfile)

    def _pop_signal(self):
        return self._signal_queue.pop(0) if self._signal_queue else None

    def _wakeup(self):
        """
        Wake up by writing to the PIPE
        """
        try:
            os.write(self._wfile, b'.')
        except IOError as e:
            if e.errno not in [errno.EAGAIN, errno.EINTR]:
                raise

    def _sleep(self, timeout=None):
        """
        Sleep until PIPE is readable or we timeout.
        A readable PIPE means a signal occurred.
        """
        try:
            ready = select.select([self._rfile], [], [], timeout)
            if not ready[0]:
                return
            while os.read(self._rfile, 1):
                pass
        except (select.error, OSError) as e:
            # TODO: select.error is a subclass of OSError since Python 3.3.
            error_number = getattr(e, 'errno', e.args[0])
            if error_number not in [errno.EAGAIN, errno.EINTR]:
                raise


def _test_signal_pipe():
    import signal
    signal_pipe = SignalPipe()
    expect = signal.SIGUSR1
    try:
        signal.signal(expect, signal_pipe.handler)
        got_none = signal_pipe.wait(timeout=0.1)
        os.kill(os.getpid(), expect)
        got = signal_pipe.wait(timeout=0.1)
    finally:
        signal_pipe.close()
    assert got_none is None, f'expect None, got={got_none}'
    assert got == expect, f'expect {expect}, got={got}'
